package cn.doitedu.demo.join;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class 时间窗口join {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射kafka中的ss-1
        tenv.executeSql("CREATE TABLE t1                  (\n" +
                "  uid BIGINT                                 \n" +
                "  ,event_id STRING                           \n" +
                "  ,action_time BIGINT                        \n" +
                "  ,rt1 as to_timestamp_ltz(action_time,3)    \n" +
                "  ,watermark for rt1 as rt1                  \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'a-1',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g1',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");


        // 建表，映射kafka中的ss-2
        tenv.executeSql("CREATE TABLE t2                  (\n" +
                "  uid BIGINT                                \n" +
                "  ,face_status int                           \n" +
                "  ,ts bigint                                 \n" +
                "  ,rt2 as to_timestamp_ltz(ts,3)             \n" +
                "  ,watermark for rt2 as rt2                  \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'a-2',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g2',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");

        tenv.executeSql(
                        "with tmp1 as (\n" +
                        "    select\n" +
                        "        window_start,\n" +
                        "        window_end,\n" +
                        "        uid,\n" +
                        "        event_id,\n" +
                        "        action_time\n" +
                        "    from table(\n" +
                        "      tumble(table t1,descriptor(rt1),interval '10' second)\n" +
                        "    )\n" +
                        "),\n" +
                        "tmp2 as (\n" +
                        "    select\n" +
                        "        window_start,\n" +
                        "        window_end,\n" +
                        "        uid,\n" +
                        "        face_status,\n" +
                        "        ts\n" +
                        "    from table(\n" +
                        "        tumble(table t2,descriptor(rt2),interval '10' second)\n" +
                        "    )\n" +
                        ")\n" +
                        "select\n" +
                        "    tmp1.window_start,\n" +
                        "    tmp1.window_end,\n" +
                        "    tmp1.uid,\n" +
                        "    tmp1.event_id,\n" +
                        "    tmp1.action_time,\n" +
                        "    tmp2.uid,\n" +
                        "    tmp2.face_status,\n" +
                        "    tmp2.ts\n" +
                        "from tmp1 \n" +
                        "join tmp2\n" +
                        "on   tmp1.window_start=tmp2.window_start\n" +
                        "and  tmp1.window_end = tmp2.window_end\n" +
                        "and  tmp1.uid = tmp2.uid  ").print();



    }


}
